package com.zf.thread_pool;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * {@link ThreadPoolExecutor} 样例
 */
public class ThreadPoolTest {
    public static Logger log = LoggerFactory.getLogger(MyThreadFactory.class);

    /**
     * 自定义线程工厂
     */
    private static class MyThreadFactory implements ThreadFactory {
        public static Logger log = LoggerFactory.getLogger(MyThreadFactory.class);

        /**
         * 记录 MyThreadFactory 线程工厂类 创建了多少个 线程工厂实例
         */
        private static final AtomicInteger poolNumber = new AtomicInteger(0);

        /**
         * 线程组
         */
        private final ThreadGroup group;

        /**
         * 记录当前线程工厂实例对象 创建了多少个线程
         */
        private final AtomicInteger threadNumber = new AtomicInteger(0);

        /**
         * 线程名前缀
         */
        private final String namePrefix;

        public MyThreadFactory() {
            // 初始化线程工厂对象的一些配置
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            namePrefix = "my-pool-" + poolNumber.incrementAndGet() + "#thread-";
        }

        /**
         * 创建线程对象
         */
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.incrementAndGet(), 0);
            t.setDaemon(false);
            t.setPriority(Thread.NORM_PRIORITY);

            // 设置线程中运行的代码出现异常时的处理逻辑
            // 线程的创建和销毁比较消耗资源，定义异常处理，可以在线程中的代码发生异常时，进行处理，防止线程被销毁
            t.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
                @Override
                public void uncaughtException(Thread t, Throwable e) {
                    // 获取异常栈的前5个信息，不足5个的取全部信息
                    StackTraceElement[] stackTrace = e.getStackTrace();
                    String errorMessage = Arrays.asList(stackTrace).subList(0, Math.min(stackTrace.length, 5))
                            .stream().map(StackTraceElement::toString).collect(Collectors.joining("\n"));
                    errorMessage = e.toString() + "\n" + errorMessage;

                    // 打印异常信息
                    log.info("线程：[{}] 异常信息：{}", t.getName(), errorMessage);
                }
            });

            return t;
        }
    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 处理器逻辑核心数
        int availableProcessors = Runtime.getRuntime().availableProcessors() - 1;

        // 工作队列
        ArrayBlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(200);

        // 创建自定义的线程工厂
        MyThreadFactory threadFactory = new MyThreadFactory();

        // 创建自定义线程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                availableProcessors, // 核心线程数
                availableProcessors * 4,// 最大线程数
                60, TimeUnit.SECONDS, // 超过核心线程数量的其他的线程，在空闲超过此时间后，线程被销毁
                workQueue,  // 工作队列，当 核心线程数 全部被占用后，新的任务被提交到工作队列；当工作队列被占满后，新提交的任务会尝试“获取已有/创建新的”线程
                threadFactory, // 线程工厂
                new ThreadPoolExecutor.CallerRunsPolicy() //拒绝策略，当工作队列被占满后，且 “最大线程数量”的所有线程都被占用时，则执行拒绝策略；CallerRunsPolicy 表示由提交任务的线程执行该任务
        );
        // execute 提交任务，不能有返回结果
        executor.execute(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            log.info("{}：正常的 execute() 的任务", Thread.currentThread());
        });


        // submit 提交任务，可以获取返回结果
        Future<String> future = executor.submit(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            log.info("{}：正常的 submit() 的任务", Thread.currentThread());
            return "正常的 submit() 的任务执行完毕";
        });
        log.info(future.get());

        // 使用submit提交的任务，如果出现异常，不会执行线程工厂中配置的异常处理，
        // 而是把异常封装到了返回的 Future 中，使用 Future.get() 获取时抛出异常。
        // 如果没有接收返回结果 或者 没有调用  Future.get() ，则异常信息丢失。
        Future<?> futureExp = executor.submit(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            throw new RuntimeException("异常 submit() 任务");
        });
        try {
            futureExp.get(); // 不调用 get() 方法，则异常信息丢失
        } catch (Exception e) {
            log.error("", e);
        }


        // 使用execute提交的任务，如果出现异常，则会执行线程工厂中配置的异常处理
        executor.execute(() -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            throw new RuntimeException("异常 execute() 任务");
        });

        // 关闭线程池，将线程池的shutdown标识设置为 true, 不再接收新的任务，继续完成已提交的任务，待所有的任务都完成后 线程池变成了 TERMINATED 状态。
        // executor.shutdown()不会阻塞当前调用线程
        executor.shutdown();
        System.out.println("执行了shutdown");


        try {
            // 超时等待60s, 如果60s内的任一时刻线程池变成了 TERMINATED 状态，立即返回true,
            // 如果60s后还不是 TERMINATED 状态则返回false
            boolean terminated = executor.awaitTermination(60, TimeUnit.SECONDS);
            if (terminated) {
                log.info("所有任务都已经完成");
            } else {
                log.info("还有任务在执行中");
            }
        } catch (InterruptedException e) {
            System.out.println("当前线程被中断了，即执行的shutdown操作的线程被停止了");
            // 由于抛出了 InterruptedException 异常，当前线程的中断状态被重置为false了，导致当前线程的中断状态消失了
            // 需要恢复中断状态
            Thread.currentThread().interrupt();
        }
        log.info("主线程结束了");

    }

}
